Live Runs and Thread Continuation

Run Live with Popen (Sync)

run_live() forces JSON mode so events can be parsed immediately. You can iterate events, monitor status, and collect the final result with wait().

from codex_local_sdk import CodexExecRequest, CodexLocalClient, SandboxMode

client = CodexLocalClient()
live = client.run_live(
    CodexExecRequest(
        prompt="Analyze architecture and propose migration plan.",
        sandbox=SandboxMode.READ_ONLY,
    )
)

for event in live.iter_events():
    print(event.type)

result = live.wait(timeout=120)
print(result.turn_status)
print(result.final_message)

Run Live with asyncio (Async)

run_live_async() returns AsyncCodexLiveRun with async iteration and async wait/result methods.

import asyncio
from codex_local_sdk import CodexExecRequest, CodexLocalClient


async def main() -> None:
    client = CodexLocalClient()
    live = await client.run_live_async(
        CodexExecRequest(prompt="Stream architecture review", json_output=True)
    )

    async for event in live.iter_events():
        print(event.type)

    final = await live.wait(timeout=120)
    print(final.turn_status, final.final_message)


asyncio.run(main())

Thread Sessions for Multi-turn Flows

start_thread() captures thread ID and returns CodexThreadSession. Add session_name to persist it via your configured session store.

from codex_local_sdk import CodexLocalClient, JsonFileSessionStore, SandboxMode

client = CodexLocalClient(session_store=JsonFileSessionStore(".codex_sessions.json"))
session, first = client.start_thread(
    prompt="Review repository and create a 3-phase plan.",
    sandbox=SandboxMode.READ_ONLY,
    session_name="repo-plan",
)

print(session.session_id)
print(first.is_turn_completed)

second = session.continue_prompt("Implement phase 1 tasks", json_output=True)
print(second.final_message)

Resume Patterns

Resume by Session ID

result = client.resume(
    prompt="Continue and write test checklist.",
    session_id="thread-abc123",
    json_output=True,
    last=False,
)
print(result.turn_status)

Resume by Session Name (Persistent Store)

result = client.resume(
    prompt="Continue and write implementation tasks.",
    session_name="repo-plan",
    last=False,
    json_output=True,
)

Async Resume Live

live = await client.resume_live_async(
    prompt="Continue and stream reasoning summary.",
    session_name="repo-plan",
    last=False,
)
async for event in live.iter_events():
    print(event.type)
result = await live.wait()

Completion and Monitoring Checks

Use turn status helpers and return codes to branch orchestration logic.

live = client.run_live(CodexExecRequest(prompt="Long analysis", json_output=True))

for event in live.iter_events():
    if event.type in {"turn.completed", "turn.failed", "turn.interrupted"}:
        print("terminal event", event.type)

result = live.wait()
if result.is_turn_terminal and result.ok:
    print("done", result.turn_status)
else:
    print("handle failure", result.return_code)

Retry Strategy for Follow-up Turns

Retry policy is configured on CodexLocalClient and applies to sync resume calls. Async wrappers reuse the same retry policy semantics, and live methods retry startup failures before returning a handle.